package com.example.rxtest;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

/**
 * 作者 fengxn
 * 日期 2018/2/27
 * 功能描述：
 */

public class RxJavaTest
{
    public void test()
    {

        //可以理解为   View设置onClick事件
//          View ----->Observable
//          view.setOnclick() --->subscribe
//          observer  --- OnClickListener 事件

        //第一步  创建被观察者  Observable
        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>()
        {
            @Override
            public void call(Subscriber<? super String> subscriber)
            {
                //第四步  执行回调
                subscriber.onNext("呵呵");
                subscriber.onCompleted();
                subscriber.onError(new RuntimeException("11111"));

            }
        });

//        第二步 创建观察者
        Observer<String> observer = new Observer<String>()
        {
            @Override
            public void onCompleted()
            {
                System.out.println("onCompleted");
            }

            @Override
            public void onError(Throwable e)
            {
                System.out.println("onError");
            }

            @Override
            public void onNext(String s)
            {
                System.out.println(s);
            }
        };
        //第三步  订阅
        observable.subscribe(observer);
    }

    public void just()
    {

        Observable<User> observable = Observable.just(new User("1", "黑恶虎"), new User("2", "不好找"));

        observable.subscribe(new Observer<User>()
        {

            @Override
            public void onCompleted()
            {

            }

            @Override
            public void onError(Throwable e)
            {

            }

            @Override
            public void onNext(User user)
            {
                System.out.println(user.getName());
            }
        });
    }

    public void form()
    {
        ArrayList<String> list = new ArrayList<>();
        for (int i = 0; i < 10; i++)
        {
            list.add("" + i);
        }

        Observable<String> observable = Observable.from(list);
        observable.subscribe(new Subscriber<String>()
        {
            @Override
            public void onCompleted()
            {

            }

            @Override
            public void onError(Throwable e)
            {

            }

            @Override
            public void onNext(String s)
            {
                System.out.println(s);
            }
        });

    }

    /**
     * 间隔发送
     */
    public void interval()
    {
        /**
         * 参数一 延迟 1秒
         * 参数二 间隔时间
         * 参数三 间隔时间的单位
         *
         *
         */
        Observable<Long> ob = Observable.interval(1000, 1000, TimeUnit.SECONDS);
        ob.subscribe(new Subscriber<Long>()
        {
            @Override
            public void onCompleted()
            {

            }

            @Override
            public void onError(Throwable e)
            {

            }

            @Override
            public void onNext(Long aLong)
            {

            }
        });


    }

    /**
     * 定时器
     */
    public void timer()
    {
        Observable.timer(1000, TimeUnit.SECONDS).subscribe(new Subscriber<Long>()
        {
            @Override
            public void onCompleted()
            {

            }

            @Override
            public void onError(Throwable e)
            {

            }

            @Override
            public void onNext(Long aLong)
            {

            }
        });
    }

    //---------------------------转化--------------------


    public void map()
    {
        //一对一转化
        Observable<String> ob = Observable.just("1");
        ob.map(new Func1<String, Integer>()
        {
            @Override
            public Integer call(String s)
            {
                return Integer.valueOf(s);
            }
        }).subscribe(new Subscriber<Integer>()
        {
            @Override
            public void onCompleted()
            {
                System.out.println("onCompleted");
            }

            @Override
            public void onError(Throwable e)
            {
                System.out.println("onError");
            }

            @Override
            public void onNext(Integer integer)
            {
                System.out.println(integer);
            }
        });


    }

    public void flatMap()
    {
        String titles[] = {"1", "2", "3"};
        Observable.from(titles).flatMap(new Func1<String, Observable<Integer>>()
        {
            @Override
            public Observable<Integer> call(String s)
            {
                return Observable.just(Integer.valueOf(s));
            }
        }).subscribe(new Subscriber<Integer>()
        {
            @Override
            public void onCompleted()
            {

            }

            @Override
            public void onError(Throwable e)
            {

            }

            @Override
            public void onNext(Integer integer)
            {
                System.out.println(integer);

            }
        });


    }

    /**
     *
     */
    public void buffer()
    {


    }

    /**
     *
     *
     */
    public void merge()
    {
        Observable<String> ob1 = Observable.from(new String[]{"1", "2", "4"});
        Observable<String> ob2 = Observable.from(new String[]{"5", "6", "6", "9"});

        Observable.merge(ob1, ob2).subscribe(new Subscriber<String>()
        {
            @Override
            public void onCompleted()
            {

            }

            @Override
            public void onError(Throwable e)
            {

            }

            @Override
            public void onNext(String s)
            {
                System.out.println(s);
            }
        });
    }


    public void zip()
    {
        Integer[] i1 = {1, 12, 4};
        Integer[] i2 = {2, 13, 16, 20};

        Observable<Integer> observable1 = Observable.from(i1).observeOn(Schedulers.newThread());
        Observable<Integer> observable2 = Observable.from(i2).observeOn(Schedulers.newThread());
        Observable.zip(observable1, observable2, new Func2<Integer, Integer, Integer>()
        {
            @Override
            public Integer call(Integer integer, Integer integer2)
            {
                return integer + integer2;
            }
        }).subscribeOn(Schedulers.io())

                .subscribe(new Subscriber<Integer>()
                {
                    @Override

                    public void onCompleted()
                    {

                    }

                    @Override
                    public void onError(Throwable e)
                    {

                    }

                    @Override
                    public void onNext(Integer integer)
                    {
                        System.out.println(integer);
                    }
                });


    }

}
